package com.sc.nft.admin.consume;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.lang.Assert;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sc.nft.dao.ExportRecordsDao;
import com.sc.nft.dao.TempDaoUserCollectionHoldDao;
import com.sc.nft.dao.UserCollectionDao;
import com.sc.nft.dao.UserInfoDao;
import com.sc.nft.dto.TeamHoldCollectionVO;
import com.sc.nft.entity.ExportRecords;
import com.sc.nft.entity.TempDaoUserCollectionHold;
import com.sc.nft.entity.UserInfo;
import com.sc.nft.entity.dto.DaoTeamHoldCollectionDTO;
import com.sc.nft.enums.ExportTypeStatusEnum;
import com.sc.nft.enums.MessageTopic;
import com.sc.nft.enums.RedisKeyEnum;
import com.sc.nft.exception.GlobalRunTimeException;
import com.sc.nft.sup.ErrorCode;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author 刘国航
 * @desc
 * @since 2024/10/31 14:04
 */
@RocketMQMessageListener(topic = MessageTopic.DAO_TEAM_HOLD_COLLECTION + "${spring.profiles.active}", consumerGroup = MessageTopic.DAO_TEAM_HOLD_COLLECTION + "${spring.profiles.active}")
@Component
@Slf4j
@RequiredArgsConstructor
public class DaoTeamHoldCollectionConsumer implements RocketMQListener<DaoTeamHoldCollectionDTO> {

    private final ExportRecordsDao exportRecordsDao;
    private final Redisson redisson;
    private final UserCollectionDao userCollectionDao;
    private final UserInfoDao userInfoDao;
    private final TempDaoUserCollectionHoldDao tempDaoUserCollectionHoldDao;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void onMessage(DaoTeamHoldCollectionDTO daoTeamHoldCollectionDTO) {
        Long exportRecordsId = daoTeamHoldCollectionDTO.getExportRecordsId();
        String key = RedisKeyEnum.DAO_TEAM_HOLD_COLLECTION.getKey() + exportRecordsId;
        RLock lock = redisson.getLock(key);
        try {
            //尝试数据加锁
            if (!lock.tryLock(10, TimeUnit.SECONDS)) {
                throw new GlobalRunTimeException("操作频繁，请稍后重试");
            }
            ExportRecords exportRecords = exportRecordsDao.getById(exportRecordsId);
            Assert.notNull(exportRecords, () -> new GlobalRunTimeException("导出记录不存在"));
            Assert.isTrue(Objects.equals(exportRecords.getStatus(), ExportTypeStatusEnum.EXPORT_STATUS_IN.getCode()), () -> new GlobalRunTimeException("导出任务已结束"));

            List<TeamHoldCollectionVO> data = userCollectionDao.daoTeamHoldCollection(daoTeamHoldCollectionDTO.getDaoUserId(), daoTeamHoldCollectionDTO.getCollectionId(), daoTeamHoldCollectionDTO.getUserTel());
            if (CollectionUtil.isEmpty(data)) {
                exportRecords.setStatus(ExportTypeStatusEnum.EXPORT_STATUS_FAIL.getCode());
                if (!exportRecordsDao.updateById(exportRecords)) {
                    throw new GlobalRunTimeException(ErrorCode.UPDATE_ERROR);
                }
                return;
            }

            List<Long> userIdList = data.stream().map(TeamHoldCollectionVO::getUserId).distinct().collect(Collectors.toList());
            List<UserInfo> userInfoList = userInfoDao.getByIdList(userIdList);
            Map<Long, UserInfo> userInfoMap = CollectionUtil.isNotEmpty(userInfoList) ? userInfoList.stream().collect(Collectors.toMap(UserInfo::getId, Function.identity())) : Maps.newHashMap();

            List<TempDaoUserCollectionHold> tempDaoUserCollectionHoldList = Lists.newArrayList();
            data.forEach(v -> {
                TempDaoUserCollectionHold tempDaoUserCollectionHold = new TempDaoUserCollectionHold();
                tempDaoUserCollectionHold.setExportRecordsId(exportRecordsId);
                UserInfo userInfo = userInfoMap.get(v.getUserId());
                tempDaoUserCollectionHold.setUserTel(Objects.nonNull(userInfo) ? userInfo.getUserTel() : "");
                tempDaoUserCollectionHold.setNickName(Objects.nonNull(userInfo) ? userInfo.getNickName() : "");
                tempDaoUserCollectionHold.setCollectionName(v.getCollectionName());
                tempDaoUserCollectionHold.setHoldCount(v.getHoldCount());
                tempDaoUserCollectionHoldList.add(tempDaoUserCollectionHold);
            });

            tempDaoUserCollectionHoldDao.saveBatch(tempDaoUserCollectionHoldList);

            exportRecords.setStatus(ExportTypeStatusEnum.EXPORT_STATUS_SUCCESS.getCode());
            if (!exportRecordsDao.updateById(exportRecords)) {
                throw new GlobalRunTimeException(ErrorCode.UPDATE_ERROR);
            }
        } catch (Exception e) {
            log.error("", e);
            throw new GlobalRunTimeException(e.getMessage());
        } finally {
            if (lock.isLocked() && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }


    }

}
